package com.rem.flink.flink6ProcessFunction;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * ProcessFunction 最基本的处理函数，基于 DataStream 直接调用.process()时作为参数传入。
 * KeyedProcessFunction 对流按键分区后的处理函数，基于 KeyedStream 调用.process()时作为参数传入。要想使用定时器，比如基于 KeyedStream。
 * ProcessWindowFunction 开窗之后的处理函数，也是全窗口函数的代表。基于 WindowedStream 调用.process()时作为参数传入。
 * ProcessAllWindowFunction 同样是开窗之后的处理函数，基于 AllWindowedStream 调用.process()时作为参数传入。
 * CoProcessFunction 合并（connect）两条流之后的处理函数，基于 ConnectedStreams 调用.process()时作为参数传入。
 * ProcessJoinFunction 间隔连接（interval join）两条流之后的处理函数，基于 IntervalJoined 调用.process()时作为参数传入
 * BroadcastProcessFunction 广播连接流处理函数，基于 BroadcastConnectedStream 调用.process()时作为参数传入。是一个未keyBy 的普通 DataStream 与一个广播流（BroadcastStream）做连接（conncet）之后的产物
 * KeyedBroadcastProcessFunction 按键分区的广播连接流处理函数，同样是基于 BroadcastConnectedStream 调用.process()时作为参数传入。与 BroadcastProcessFunction 不同的是，这时的广播连接流，是一个 KeyedStream与广播流（BroadcastStream）做连接之后的产物。
 * <p>
 * <p>
 * ProcessAllWindowFunctionTest
 * 并行度为1情况下 统计热门url windowALL 单分区情况下
 *
 * @author Rem
 * @date 2022-10-13
 */

public class ProcessAllWindowFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.  <Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));

        /**
         * 统计
         *
         */
        stream.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new ProcessAllWindowFunction<>() {
                    @Override
                    public void process(ProcessAllWindowFunction<Event, Object, TimeWindow>.Context context, Iterable<Event> elements, Collector<Object> out) throws Exception {
                        HashMap<String, Long> pvMap = new HashMap<>();

                        for (Event event : elements) {
                            String url = event.getUrl();
                            //url出现的次数
                            if (pvMap.containsKey(url)) {
                                //存在 在原有基础上加1
                                pvMap.put(url, pvMap.get(url) + 1L);
                            } else {
                                //不存在加1
                                pvMap.put(url, 1L);
                            }

                        }

                        List<Tuple2<String, Long>> pvList = new ArrayList<>();
                        pvMap.keySet().forEach(k -> pvList.add(Tuple2.of(k, pvMap.get(k))));
                        pvList.sort((o1, o2) -> (int) (o2.f1 - o1.f1));

                        StringBuilder result = new StringBuilder();
                        result.append("========================================\n");
                        for (int i = 0; i < pvList.size(); i++) {
                            Tuple2<String, Long> temp = pvList.get(i);
                            String info = "浏览量No." + (i + 1) +
                                    " url：" + temp.f0 +
                                    " pv量：" + temp.f1 +
                                    " 窗口结束时间：" + new Timestamp(context.window().getEnd()) + "\n";
                            result.append(info);
                        }
                        result.append("========================================\n");
                        out.collect(result);

                    }
                }).print();

        env.execute();


    }


}
